package com.ibm.cps.storm;

import java.util.Map;

import org.apache.log4j.Logger;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import com.google.common.base.Throwables;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.cps.newmessage.SerializableDataSourceMetadata;
import com.ibm.factories.DataSourceFactory;
import com.ibm.util.TopologyMessageFieldIds;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;

/**
 * store data source metadata and parse raw data message
 * @author Jingjing Wang
 *
 * 2015-7-13
 */
public class DataSourceBolt extends BaseRichBolt {

	private static final long serialVersionUID = 1L;
	private DataSourceFactory dataSourceFactory;
	private static final Logger logger = Logger.getLogger(DataSourceBolt.class);
	private OutputCollector collector;

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		dataSourceFactory = new DataSourceFactory();
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		try {
			String sourceStream = input.getSourceStreamId();
			switch (sourceStream) {
			case TopologyStreamIds.DATASOURCE_SPOUT_STREAM:
				SerializableDataSourceMetadata metadata = (SerializableDataSourceMetadata) input
						.getValue(0);
				dataSourceFactory.addDataSource(metadata);
				break;

			case TopologyStreamIds.DATASOURCE_DELETE_SPOUT_STREAM:
				SerializableDataSourceMetadata metadata1 = (SerializableDataSourceMetadata) input
						.getValue(0);
				dataSourceFactory.deleteDataSource(metadata1);
				break;

			case TopologyStreamIds.RAW_DATA_STREAM:
				RawDataMessage rawDataMessage = (RawDataMessage) input
						.getValue(0);
				ParsedDataMessage parsedData = dataSourceFactory
						.parseRawData(rawDataMessage);
				collector.emit(TopologyStreamIds.PARSED_DATA_STREAM,
						new Values(parsedData, rawDataMessage.getTsKey(),
								rawDataMessage.getSourceid()));
				break;

			default:
				break;
			}
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declareStream(TopologyStreamIds.PARSED_DATA_STREAM,
				new Fields(TopologyMessageFieldIds.MESSAGE,
						TopologyMessageFieldIds.TS_KEY,
						TopologyMessageFieldIds.SOURCE));
	}

}
